package com.scut.chenlujie.library.clink.impl;

import com.scut.chenlujie.library.clink.core.IoProvider;
import com.scut.chenlujie.library.clink.utils.CloseUtils;

import java.io.IOException;
import java.nio.channels.*;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

public class IoSelectorProvider implements IoProvider {
    private final AtomicBoolean isClosed = new AtomicBoolean(false);

    // 是否处于某个过程
    private final AtomicBoolean inRegInput = new AtomicBoolean(false);
    private final AtomicBoolean inRegOutput = new AtomicBoolean(false);

    //读写两个selector分开
    private final Selector readSelector;
    private final Selector writeSelector;

    //放到达数据的channel而要触发执行任务
    private final Map<SelectionKey , Runnable> inputCallbackMap = new HashMap<>();
    private final Map<SelectionKey , Runnable> outputCallbackMap = new HashMap<>();

    //读写两个线程池 ,分别执行上面两个map中的callback任务
    private final ExecutorService inputHandlePool;
    private final ExecutorService outputHandlePool;

    public IoSelectorProvider() throws IOException {
        this.readSelector = Selector.open();
        this.writeSelector = Selector.open();
        this.inputHandlePool = Executors.newFixedThreadPool(20,
                new NameableThreadFactory("IoProvider-Input-Thread-"));
        this.outputHandlePool = Executors.newFixedThreadPool(20,
                new NameableThreadFactory("IoProvider-Output-Thread-"));

        // 开始输出输入的监听
        startRead();
        startWrite();

    }



    private static SelectionKey registerSelection(SocketChannel channel, Selector selector,
                                                  int registerOps, AtomicBoolean locker,
                                                  Map<SelectionKey, Runnable> map,
                                                  Runnable runnable) {

        
        synchronized (locker) {
            // 设置锁定状态
            locker.set(true);

            try {
                // 唤醒当前的selector，让selector不处于select()状态
                selector.wakeup();

                SelectionKey key = null;
                if (channel.isRegistered()) {
                    // 查询是否已经注册过
                    key = channel.keyFor(selector);
                    if (key != null) {
                        key.interestOps(key.readyOps() | registerOps);
                    }
                }

                if (key == null) {
                    // 注册selector得到Key
                    key = channel.register(selector, registerOps);
                    // 注册回调
                    map.put(key, runnable);
                }

                return key;
            } catch (ClosedChannelException|CancelledKeyException|ClosedSelectorException e) {
                return null;
            } finally {
                // 解除锁定状态
                locker.set(false);
                try {
                    // 通知
                    locker.notify();
                } catch (Exception ignored) {
                }
            }
        }
    }

    private void unregisterSelection(SocketChannel channel ,Selector selector ,Map<SelectionKey , Runnable> map ,AtomicBoolean locker){
        //其他线程去unregister
        synchronized (locker){
            locker.set(true);
            selector.wakeup();
            try {
                if(channel.isRegistered()){
                    SelectionKey key = channel.keyFor(selector);
                    if(key != null){
                        // 取消监听的方法
                        key.cancel();
                        map.remove(key);
                        //防止堵塞在Selector.select()的时候没有意识到更新了, 后面如果不唤醒 一直堵塞在select那里 就有可能收到删掉的key而触发下面
                        //selector.wakeup();
                    }
                }
            }finally {
                locker.set(false);
                try {
                    locker.notifyAll();
                }catch (Exception ignored){

                }
            }

        }

    }

    static class SelectThread extends Thread{
        private final AtomicBoolean isClosed;
        private final AtomicBoolean locker;
        private final Selector selector;
        private final int keyOps;
        //放到达数据的channel而要触发执行任务
        private final Map<SelectionKey , Runnable> callbackMap;
        //读写两个线程池 ,分别执行上面两个map中的callback任务
        private final ExecutorService pool;

        SelectThread(String name, AtomicBoolean isClosed, AtomicBoolean locker, Selector selector, int keyOps, Map<SelectionKey, Runnable> callbackMap, ExecutorService pool) {
            super(name);
            this.isClosed = isClosed;
            this.locker = locker;
            this.selector = selector;
            this.keyOps = keyOps;
            this.callbackMap = callbackMap;
            this.pool = pool;
            this.setPriority(Thread.MAX_PRIORITY);
        }
        @Override
        public void run(){
            super.run();
            AtomicBoolean locker = this.locker;
            AtomicBoolean isClosed = this.isClosed;
            Selector selector = this.selector;
            Map<SelectionKey,Runnable> callbackMap = this.callbackMap;
            ExecutorService pool = this.pool;
            int keyOps = this.keyOps;

            while(!isClosed.get()){
                try {
                    if(selector.select() == 0){
                        waitSelection(locker);
                        continue;
                    }else if(locker.get()){
                        waitSelection(locker);
                    }
                    //有数据的channel 返回到Set<SelectionKey>中
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()){
                        SelectionKey selectionKey = iterator.next();
                        if(selectionKey.isValid()){
                            handleSelection(selectionKey , keyOps , callbackMap , pool ,locker);
                        }
                        iterator.remove();
                    }
                    selectionKeys.clear();

                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        }
    }

    private  void startRead(){
        Thread thread = new SelectThread("Clink IoSelectorProvider ReadSelector Thread" , isClosed , inRegInput ,readSelector ,SelectionKey.OP_READ , inputCallbackMap , inputHandlePool);
        thread.start();
    }

    private  void startWrite(){
        Thread thread = new SelectThread("Clink IoSelectorProvider WriteSelector Thread" , isClosed , inRegOutput ,writeSelector ,SelectionKey.OP_WRITE, outputCallbackMap , outputHandlePool);
        thread.start();
    }

//    private  void startRead() {
//        Thread thread = new Thread("Clink IoSelectorProvider ReadSelector Thread"){
//            @Override
//            public void run(){
//                AtomicBoolean locker = inRegInput;
//                while(!isClosed.get()){
//                    try {
//
//                        if(readSelector.select() == 0){
//                        waitSelection(inRegInput);
//                    }
//                    //select()被唤醒可能有返回已经扫描过并且有数据的通道
//                    else if (locker.get()){
//                        waitSelection(inRegInput);
//                    }
//                    //有数据的channel 返回到Set<SelectionKey>中
//                    Set<SelectionKey> selectionKeys = readSelector.selectedKeys();
//
//                    //A key is valid upon creation and remains so until it is cancelled, its channel is closed, or its selector is closed.
//                        // for循环可能会出现问题 -> 当前的key已经被删除掉了 会引发异常 所以说正确的做法应该是替换成一个迭代器
////                    for (SelectionKey selectionKey : selectionKeys) {
////                        if(selectionKey.isValid()){
////                            //异步交给线程池去做 如果线程池没有马上读取数据的话 那么下次select()还是会把任务放入线程池，于是有太多的冗余任务了
////                            handleSelection(selectionKey , SelectionKey.OP_READ , inputCallbackMap , inputHandlePool );
////                        }
////                    }
////
//                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
//                    while (iterator.hasNext()){
//                        SelectionKey selectionKey = iterator.next();
//                        if(selectionKey.isValid()){
//                            handleSelection(selectionKey , SelectionKey.OP_READ , inputCallbackMap , inputHandlePool );
//                        }
//                        iterator.remove();
//                    }
//                    selectionKeys.clear();
//                    }catch (IOException e){
//                        e.printStackTrace();
//                    }
//                }
//            }
//        };
//        thread.setPriority(Thread.MAX_PRIORITY);
//        thread.start();
//    }

//    private void startWrite() {
//        Thread thread = new Thread("Clink IoSelectorProvider WriteSeletor Thread"){
//            @Override
//            public void run(){
//                while(!isClosed.get()){
//                    try {
//                        if(writeSelector.select() == 0){
//                            waitSelection(inRegOutput);
//                            continue;
//                        }
//                        //有数据的channel 返回到Set<SelectionKey>中
//                        Set<SelectionKey> selectionKeys = writeSelector.selectedKeys();
//                        for (SelectionKey selectionKey : selectionKeys) {
//                            if(selectionKey.isValid()){
//                                //异步交给线程池去做 如果线程池没有马上读取数据的话 那么下次select()还是会把任务放入线程池，于是有太多的冗余任务了
//                                handleSelection(selectionKey , SelectionKey.OP_WRITE , outputCallbackMap , outputHandlePool );
//                            }
//                        }
//                        selectionKeys.clear();
//                    }catch (IOException e){
//                        e.printStackTrace();
//                    }
//                }
//            }
//        };
//        thread.setPriority(Thread.MAX_PRIORITY);
//        thread.start();
//    }

    //等待注册监听完成之后再下一次select(),否则wakeup之后又马上select(),新的还没注册进去
    private static void waitSelection(final AtomicBoolean lock){
        synchronized (lock){
            try {
                if(lock.get()){
                    lock.wait();
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }


    private static void handleSelection(SelectionKey key, int keyOps, Map<SelectionKey, Runnable> CallbackMap, ExecutorService HandlePool ,AtomicBoolean locker) {
        //重点！！！！
        //取消对keyOps的监听
        synchronized (locker){
            try{
                key.interestOps(key.readyOps() & ~keyOps); //因为readselector和writeselector是分开的，所以此处等于key.cancel();
            }catch (CancelledKeyException e){
                return ;
            }

        }

        Runnable runnable = null;
        try {
            runnable = CallbackMap.get(key);
        }catch (Exception ignored){

        }
        //可能在中间时刻取消了某个key，unregister监听某个通道
        if(runnable != null && !HandlePool.isShutdown()){
            //异步调度
            HandlePool.execute(runnable);
        }

    }

    @Override
    public boolean registerInput(SocketChannel channel, HandleProviderCallback callback) throws ClosedChannelException {
        return registerSelection( channel,readSelector ,SelectionKey.OP_READ,inRegInput , inputCallbackMap, callback) != null ;

    }

    @Override
    public boolean registerOutput(SocketChannel channel, HandleProviderCallback callback) {
        return registerSelection(channel ,writeSelector ,SelectionKey.OP_WRITE ,inRegOutput ,outputCallbackMap ,callback) != null;
    }

    @Override
    public void unregisterInput(SocketChannel channel) {
        unregisterSelection(channel , readSelector , inputCallbackMap , inRegInput);
    }

    @Override
    public void unregisterOutput(SocketChannel channel) {
        unregisterSelection(channel ,writeSelector , outputCallbackMap  ,inRegOutput);
    }

    @Override
    public void close() throws IOException {
        if(isClosed.compareAndSet(false ,true)){
            inputHandlePool.shutdown();
            outputHandlePool.shutdown();

            inputCallbackMap.clear();
            outputCallbackMap.clear();
            //close 自带 wakeup
//            readSelector.wakeup();
//            writeSelector.wakeup();

            CloseUtils.close(readSelector, writeSelector);
        }
    }


}
